package com.hao.chapter11;

import org.apache.flink.table.api.EnvironmentSettings;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;

import static org.apache.flink.table.api.Expressions.$;

public class CreateTableTest {
    public static void main(String[] args) {

        //1.定义流环境配置来创建表执行环境
        EnvironmentSettings streamSettings = EnvironmentSettings.newInstance()
                .inStreamingMode() //流环境
                .useBlinkPlanner()
                .build();

        TableEnvironment tableEnv = TableEnvironment.create(streamSettings);

        //创建输入表
        String createInputDDL = "CREATE TABLE clickTable (" +
                " user_name STRING, " +
                " url STRING, " +
                " ts BIGINT " +
                ") WITH (" +
                "'connector'= 'filesystem'," + //连接器为文件流
                "'path' = 'input/clicks.txt'," + //路径
                " 'format' =  'csv' " + //格式
                ")";

        //执行
        tableEnv.executeSql(createInputDDL);


        //创建输出表
        String createOutDDL = "CREATE TABLE outTable (" +
                " user_name STRING, " +
                " url STRING " +
                ") WITH (" +
                " 'connector' = 'filesystem'," + //连接器为文件流
                " 'path' = 'output'," + //路径
                " 'format' =  'csv' " + //格式
                ")";
        //执行
        tableEnv.executeSql(createOutDDL);

    }
}
